package com.own.business.sync.trigger.store.polling.core;

import com.alibaba.fastjson2.JSON;
import com.own.business.sync.trigger.base.bak.util.SyncTriggerBakUtil;
import com.own.business.sync.trigger.base.config.SyncTriggerPollingCondition;
import com.own.business.sync.trigger.base.entity.SyncTrigger;
import com.own.business.sync.trigger.base.method.ResolveMethodSpringHook;
import com.own.business.sync.trigger.base.property.SyncTriggerProperty;
import com.own.component.common.util.executor.ExecutorUtil;
import com.own.component.store.core.Store;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.context.annotation.Conditional;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;

/**
 * SyncTriggerStorePollingTask
 * <p>
 * 启动完成后开始执行
 *
 * @author chenxueli
 * @date 2024-01-08 15:48:43
 */
@Slf4j
@Component
@Conditional(SyncTriggerPollingCondition.class)
public class SyncTriggerStorePollingTask {

    private final Object lock = new Object();
    @Resource
    private SyncTriggerProperty syncTriggerProperty;
    @Resource
    private SyncTriggerBakUtil syncTriggerBakUtil;
    @Resource
    private ResolveMethodSpringHook resolveMethodSpringHook;
    @Resource
    private ExecutorUtil executorUtil;
    @Resource
    private Store<String> store;

    @PostConstruct
    public void init() {
        executorUtil.execute(this::polling);
    }

    /**
     * 轮询信息
     */
    @SuppressWarnings("all")
    private void polling() {
        var key = syncTriggerProperty.getStoreKey();
        var max = syncTriggerProperty.getPollingTimeList().length - 1;
        var service = ExecutorUtil.newVirtualThreadPerTaskExecutor();
        // 采用while true的方式可以避免使用递归方式造成的栈溢出的问题
        var c = 0;
        while (true) {
            synchronized (lock) {
                // 从缓存中获取数据
                var item = store.list().leftPop(key);
                if (StringUtils.isNotBlank(item)) {
                    var data = JSON.parseObject(item, SyncTrigger.class);
                    var method = resolveMethodSpringHook.getMethod(data.getModuleName());
                    if (method != null) {
                        var operation = resolveMethodSpringHook.getMethod(data.getType(), method);
                        // 使用虚拟线程进行消费
                        service.execute(() -> {
                            try {
                                operation.accept(data);
                            } catch (Exception e) {
                                syncTriggerBakUtil.save(data, e.getMessage());
                            }
                        });
                    } else {
                        syncTriggerBakUtil.save(data, "未找到处理方法");
                    }
                    log.debug("处理完成:{},{},{}", data.getModuleName(), data.getType(), data.getTargetId());
                    c = 0;
                } else {
                    var index = Math.min(c, max);
                    var timeout = syncTriggerProperty.getPollingTimeList()[index];
                    log.info("未查询到调整的数据，开始休眠" + timeout + "ms");
                    try {
                        lock.wait(timeout);
                        c++;
                    } catch (InterruptedException e) {
                        log.error("等待异常", e);
                    }
                }
            }
        }
    }

}
